package com.cike.sparkstudy.sql.java;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import scala.Tuple2;

/**
 * JDBC数据源
 */
public class JDBCDataSource {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setMaster("local").setAppName("JDBCDataSource");
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = new SQLContext(sc);

        //首先需要数据库连接字符串
        HashMap<String, String> options = new HashMap<String, String>();
        options.put("url", "jdbc:mysql://bigdata:3306/testdb");
        options.put("dbtable", "testdb.student_infos");
        options.put("user", "root");
        options.put("password", "root");
        //通过read()方法加载数据为DataFrame
        DataFrame studentInfosDF = sqlContext.read().format("jdbc").options(options).load();
        studentInfosDF.show();


        options.put("dbtable", "testdb.student_scores");
        DataFrame studentScoresDF = sqlContext.read().format("jdbc").options(options).load();
        studentScoresDF.show();

        //将两个DataFrame转换为JavaPairRDD,然后join
        JavaPairRDD<String, Tuple2<Integer, Integer>> studentsRDD =
                studentInfosDF.javaRDD().mapToPair(new PairFunction<Row, String, Integer>() {
                    @Override
                    public Tuple2<String, Integer> call(Row row) throws Exception {
                        return new Tuple2<String, Integer>(
                                row.getString(0),
                                Integer.valueOf(String.valueOf(row.get(1))));
                    }
                }).join(studentScoresDF.javaRDD().mapToPair(
                        new PairFunction<Row, String, Integer>() {
                            @Override
                            public Tuple2<String, Integer> call(Row row) throws Exception {
                                return new Tuple2<String, Integer>(String.valueOf(row.get(0)),
                                        Integer.valueOf(String.valueOf(row.get(1))));
                            }
                        }));

        //将javaPairRDD转换为JavaRDD<Row>
        JavaRDD<Row> studentRowsRDD = studentsRDD.map(
                new Function<Tuple2<String, Tuple2<Integer, Integer>>, Row>() {
                    @Override
                    public Row call(Tuple2<String, Tuple2<Integer, Integer>> tuple) throws Exception {
                        return RowFactory.create(tuple._1, tuple._2._1, tuple._2._2);
                    }
                });

        //过滤分数大于80分的数据
        JavaRDD<Row> filterStudentRowsRDD = studentRowsRDD.filter(new Function<Row, Boolean>() {
            @Override
            public Boolean call(Row row) throws Exception {
                if (row.getInt(2) > 80) {
                    return true;
                }
                return false;
            }
        });


        //创建元数据
        ArrayList<StructField> structFields = new ArrayList<StructField>();
        structFields.add(DataTypes.createStructField("name", DataTypes.StringType, true));
        structFields.add(DataTypes.createStructField("age", DataTypes.IntegerType, true));
        structFields.add(DataTypes.createStructField("score", DataTypes.IntegerType, true));
        StructType structType = DataTypes.createStructType(structFields);

        DataFrame studentsDF = sqlContext.createDataFrame(filterStudentRowsRDD, structType);
        studentsDF.show();

        //打印输出
        Row[] rows = studentsDF.collect();
        for (Row row : rows) {
            System.out.println(row);
        }
        System.out.printf("==============================================");


        //将DataFrame中的数据保存到mysql表中去
        studentsDF.javaRDD().foreach(new VoidFunction<Row>() {
            @Override
            public void call(Row row) throws Exception {
                String sql = "insert into good_student_infos values ("
                        + "'" + String.valueOf(row.getString(0)) + "',"
                        + Integer.valueOf(String.valueOf(row.get(1))) + ","
                        + Integer.valueOf(String.valueOf(row.get(2))) + ")";

                Class.forName("com.mysql.jdbc.Driver");
                Connection conn = null;
                Statement stmt = null;
                try {
                    conn = DriverManager.getConnection(
                            "jdbc:mysql://bigdata:3306/testdb", "root", "root");
                    stmt = conn.createStatement();
                    stmt.executeUpdate(sql);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    if (stmt != null) {
                        stmt.close();
                    }
                    if (conn != null) {
                        conn.close();
                    }
                }
            }
        });


    }
}
